package com.zhou;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Optional;

/**
 * @author lcz
 */
@Component
@Slf4j
public class KafkaReceiver {
    @Value("${spring.kafka.yzfar.index}")
    String yzindex;
    @Value("${spring.kafka.yzfar.code}")
    String rqcode;

    private final String timeDelay = "timeDelayVaild";

    @KafkaListener(topics = {timeDelay})
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        Date day = new Date();
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            String message = (String) kafkaMessage.get();
            if (message.startsWith("{")) {
                JSONObject jsonObject = JSONObject.parseObject(message);
                String dateStop = jsonObject.getString(yzindex);
                SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String time = df.format(day);
                try {
                    Date d2 = df.parse(dateStop);
                    long diff = d2.getTime() - day.getTime();
                    long diffSeconds = diff / 1000 % 60;
                    long diffMinutes = diff / (60 * 1000) % 60;
                    long diffHours = diff / (60 * 60 * 1000) % 24;
                    long diffDays = diff / (24 * 60 * 60 * 1000);
                    Boolean equal = diffSeconds > 0  && diffSeconds < 5 && !(diffMinutes <0) && !(diffHours<0) && !(diffDays <0);
                    if (equal) {
                        log.info("地区代码"+jsonObject.get(rqcode) + " 与本地服务器时间合格");
                    } else {
                        log.info("地区代码"+jsonObject.get(rqcode) + " 与本地服务器时间不合格!两个时间相差："+diffDays + " 天, "+diffHours + " 小时, "
                                +diffMinutes + " 分钟, "+diffSeconds + " 秒.");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                ack.acknowledge();
            }
        }

    }
}